package simple;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.HashSet;
import java.util.Set;

public class PullConsumer {
    public static void main(String[] args) throws Exception{
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("pullConsumer");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        Set<String> topics = new HashSet<>();
        topics.add("simple");
        consumer.setRegisterTopics(topics);
        consumer.start();

        while(true) {
            for (String registerTopic : consumer.getRegisterTopics()) {
                Set<MessageQueue> messageQueueSet = consumer.fetchSubscribeMessageQueues(registerTopic);
                for (MessageQueue messageQueue : messageQueueSet) {
                    long offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY);
                    if(offset<0) {
                        offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_STORE);
                    }
                    if(offset<0) {
                        offset = consumer.maxOffset(messageQueue);
                    }
                    if(offset<0) {
                        offset = 0;
                    }
                    PullResult pull = consumer.pull(messageQueue, "*", offset, 32);
                    switch (pull.getPullStatus()) {
                        case FOUND:
                            for (MessageExt messageExt : pull.getMsgFoundList()) {
                                System.out.println(new String(messageExt.getBody()));
                            }
                            consumer.updateConsumeOffset(messageQueue,pull.getNextBeginOffset());
                    }
                }
            }
        }


    }
}
